package com.biz.primus.ms.sms.service;


import com.biz.primus.base.constant.SmsStreamConstant;
import com.biz.primus.model.misc.enums.ApiExceptionType;
import com.biz.primus.model.misc.exception.ApiException;
import com.biz.primus.model.misc.exception.ProviderNotAvailableException;
import com.biz.primus.model.misc.vo.ProviderResponse;
import com.biz.primus.model.misc.vo.ShortMessageVo;
import com.biz.primus.ms.base.service.AbstractBaseService;
import com.biz.primus.ms.sms.event.ProviderNotAvailableEvent;
import com.biz.primus.ms.sms.event.SendMessageEvent;
import com.biz.primus.ms.sms.service.checker.SendChecker;
import com.biz.primus.ms.sms.service.internal.ProviderSelector;
import com.biz.primus.ms.sms.service.internal.SystemConfigService;
import com.biz.primus.ms.sms.service.model.ServiceProvider;
import com.google.common.base.Optional;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author yanweijin
 * @date 2017/5/30
 */
@Service
@Slf4j
public class ShortMessageService extends AbstractBaseService {


    @Autowired
    private ProviderSelector providerSelector;

    @Autowired
    private SystemConfigService systemConfigService;

    @Autowired
    private List<SendChecker> checkers;


    @StreamListener(SmsStreamConstant.SMS_SEND)
    public void send(ShortMessageVo sm) {
        try {
            for (SendChecker checker : checkers) {
                checker.check(sm);
            }
            ServiceProvider provider = providerSelector.findCurrentProvider();
            sendInternal(sm, provider, null);
            log.info("send message success:{}", sm);
        } catch (Exception e) {
            for (SendChecker checker : checkers) {
                checker.afterException(sm, e);
            }
            throw e;
        }
    }


    private void sendInternal(ShortMessageVo sm, ServiceProvider provider, RetryContext retryContext) {
        try {

            ProviderResponse resp = provider.send(sm);

            log.info("使用供应商[{}]发送短信,isSuccess={},extraMessage={}", provider.getProviderName(), resp.isSuccess(), resp.getOriginalMessage());
            if (!resp.isSuccess()) {
                log.warn("发送失败的短信:{}", sm);
            }
            handleSendResponse(resp);

            publishEvent(new SendMessageEvent(this, sm, provider, resp.getTaskId()));

        } catch (ProviderNotAvailableException e) {
            publishEvent(new ProviderNotAvailableEvent(this, sm, provider, e));
            log.error(e.getMessage(), e);
            if (canRetry(retryContext, provider.getProviderId())) {
                Optional<ServiceProvider> providerOptional = providerSelector.findProviderExcludeSome(retryContext.failedProviderIdSet);
                if (providerOptional.isPresent()) {
                    sendInternal(sm, providerOptional.get(), retryContext);
                } else {
                    log.error("切换供应商重试时,没有找到其他备选供应商,已经排除的供应商:{}", retryContext.failedProviderIdSet);
                    throw new ApiException(ApiExceptionType.PROVIDER_NOT_AVAILABLE);
                }
            } else {
                throw new ApiException(ApiExceptionType.PROVIDER_NOT_AVAILABLE);
            }
        }
    }


    private void handleSendResponse(ProviderResponse response) {
        if (response.isSuccess()) return;
        log.info("send message failed, original response msg = {}", response.getOriginalMessage());
        throw new ApiException(response.getApiExceptionType());
    }

    private boolean canRetry(RetryContext retryContext, String currentProviderId) {
        if (systemConfigService.isEnableAutoSwitch()) {
            if (retryContext == null) {
                retryContext = new RetryContext(currentProviderId);
            } else {
                retryContext.addCount(currentProviderId);
            }
            if (retryContext.retryCount.get() < systemConfigService.getRetryNumber()) {
                return true;
            }
        }
        return false;
    }

    class RetryContext {
        private final AtomicInteger retryCount;
        private final Set<String> failedProviderIdSet = Sets.newHashSet();

        RetryContext(String failedProviderId) {
            this.retryCount = new AtomicInteger(1);
            this.failedProviderIdSet.add(failedProviderId);
        }

        void addCount(String failedProviderId) {
            this.failedProviderIdSet.add(failedProviderId);
            this.retryCount.addAndGet(1);
        }
    }


}
